package org.example.ordermonitor;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
import org.example.ordermonitor.OrderEvent;
import org.example.ordermonitor.OrderResult;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.TimeCharacteristic;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

public class OrderTimeout {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        env.setParallelism(1);

        DataStream<OrderEvent> dataSource = env.fromCollection(Arrays.asList(
                new OrderEvent(1L, "create", 1558430842L),
                new OrderEvent(2L, "create", 1558430843L),
                new OrderEvent(2L, "pay", 1558430844L)
        ));
        KeyedStream<OrderEvent, Long> orderEventStream = dataSource.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<OrderEvent>() {
                    @Override
                    public long extractAscendingTimestamp(OrderEvent element) {
                        return element.getEventTime() * 1000;
                    }
                })
                .keyBy(new KeySelector<OrderEvent, Long>() {
                    @Override
                    public Long getKey(OrderEvent value) throws Exception {
                        return value.getOrderId();
                    }
                });
        Pattern<OrderEvent, OrderEvent> OrderPayPattern = Pattern.<OrderEvent>begin("begin").where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return value.getEventType().equals("create");
                    }
                })
                //非严格连续
                .followedBy("follow").where(new SimpleCondition<OrderEvent>() {
                    @Override
                    public boolean filter(OrderEvent value) throws Exception {
                        return value.getEventType().equals("pay");
                    }
                })
                //设定超时时间
                .within(Time.seconds(15));
        OutputTag<OrderResult> orderTimeoutOutput = new OutputTag<OrderResult>("orderTimeout"){};
        PatternStream<OrderEvent> patternStream = CEP.pattern(orderEventStream, OrderPayPattern);

        DataStream<OrderResult> completedDataStream = patternStream.select(
                //第一个参数 刚定义的超时输出的标签
                orderTimeoutOutput,
                //第二个参数 处理超时的流
                new PatternTimeoutFunction<OrderEvent, OrderResult>() {
                    @Override
                    public OrderResult timeout(Map<String, List<OrderEvent>> pattern, long timeoutTimestamp) throws Exception {
                        Long timeoutOrderId = pattern.getOrDefault("begin", null).iterator().next().getOrderId();
                        return new OrderResult(timeoutOrderId, "timeout");
                    }
                },
                //第三个参数 处理正常的流
                new PatternSelectFunction<OrderEvent, OrderResult>() {
                    @Override
                    public OrderResult select(Map<String, List<OrderEvent>> pattern) throws Exception {
                        Long payOrderId = pattern.getOrDefault("follow", null).iterator().next().getOrderId();
                        return new OrderResult(payOrderId, "success");
                    }
                });
        //正常支付
        completedDataStream.print();
//超时支付
        DataStream<OrderResult> sideOutput = ((SingleOutputStreamOperator<OrderResult>) completedDataStream).getSideOutput(orderTimeoutOutput);sideOutput.print();
        env.execute("order timeout job");
    }
}
